package com.base.core.producer.broker;


import com.base.api.Message;
import com.base.api.MessageType;
import com.base.api.exception.MessageException;
import com.base.api.exception.MessageRunTimeException;
import com.base.common.convert.GenericMessageConverter;
import com.base.common.convert.RabbitMessageConverter;
import com.base.common.serializer.Serializer;
import com.base.common.serializer.SeriallzerFactory;
import com.base.common.serializer.impl.JacksonSerializerFactory;
import com.base.core.producer.service.MessageStoreService;
import com.google.common.base.Preconditions;
import com.google.common.base.Splitter;
import com.google.common.collect.Maps;
import lombok.extern.apachecommons.CommonsLog;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;

/**
 * 池化RabbitTemplate封装
 * 每一个topic对应一个RabbitTemplate,
 * 1、提高发送的效率
 * 2、可以根据不通的需求指定化不同的RabbitTemplate，比如每一个topic都有自己的routeKey规则
 */
@Component
@Slf4j
public class RabbitTemplateConatiner implements  RabbitTemplate.ConfirmCallback {

    private Map<String/*topic**/,RabbitTemplate> rabbitTemplateMap= Maps.newConcurrentMap();

    SeriallzerFactory serializerFactory= JacksonSerializerFactory.INSTANCE;

    private Splitter splitter=Splitter.on("#");

    private MessageStoreService messageStoreService;

    @Autowired
    private ConnectionFactory connectionFactory;


    public RabbitTemplateConatiner(MessageStoreService messageStoreService) {
        this.messageStoreService = messageStoreService;
    }

    public RabbitTemplate getTemplate(Message message) throws MessageRunTimeException {
        Preconditions.checkNotNull(message);
        String topic=message.getTopic();
        RabbitTemplate rabbitTemplate=rabbitTemplateMap.get(topic);
        if(rabbitTemplate!=null){
            return rabbitTemplate;
        }
        log.info("#RabbitTemplteConatiner.getTemplate# topic:{} is not exists,create one",topic);
        RabbitTemplate newTemplate=new RabbitTemplate(connectionFactory);
        newTemplate.setExchange(topic);
        newTemplate.setRoutingKey(message.getRoutingKey());
        newTemplate.setRetryTemplate(new RetryTemplate());
        //做提高性能的序列化，对于message的序列化
        GenericMessageConverter gmc=new GenericMessageConverter(serializerFactory.create());
        RabbitMessageConverter rmc=new RabbitMessageConverter(gmc);
        newTemplate.setMessageConverter(rmc);

        String messateType=message.getMessageType();
        if(!MessageType.RAPID.equals(messateType)){
            //不是迅速信息
            newTemplate.setConfirmCallback(this);
        }
        rabbitTemplateMap.putIfAbsent(topic,newTemplate);
        return rabbitTemplateMap.get(topic);
    }


    /**
     * 无论是confrim消息还是reliant消息，发送消息以后broker都会回调confrim
     * @param correlationData
     * @param ack
     * @param cause
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        //具体的消息应答,按照发送的时候截取
        //         CorrelationData correlationData=new CorrelationData(String.format("%s#%s",
        //                        message.getMessageId(),
        //                        System.currentTimeMillis()));
        List<String> strings=splitter.splitToList(correlationData.getId());
        String messageId=strings.get(0);
        Long sedTime=Long.parseLong(strings.get(1));
        String messageType=strings.get(2);
        if(ack){
            if(MessageType.RELIANT.endsWith(messageType)) {
                //可靠性消息
                //	当Broker 返回ACK成功时, 就是更新一下日志表里对应的消息发送状态为 SEND_OK
                messageStoreService.success(messageId);
            }
            log.info("send message is ok,confrim messageId:{},sendTime:{}",messageId,sedTime);
        }else{
            log.error("send message is Fail,confrim messageId:{},sendTime:{}",messageId,sedTime);
        }
    }
}
